package com.atguigu.Func;

import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.util.DruidDSUtil_hjy;
import com.atguigu.util.ThreadPoolUtil;
import com.atguigu.util.ThreadPoolUtil_lj;
import lombok.SneakyThrows;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.util.Collections;
import java.util.concurrent.ThreadPoolExecutor;


/**
 * @author hjy
 * @create 2023/3/27 10:11
 */
public abstract class AsyncDimFunction<T> extends RichAsyncFunction<T,T> implements AsyncJoinFunction<T>{
    private String tableName;
    private ThreadPoolExecutor threadPoolExecutor;

    public AsyncDimFunction(String tableName) {
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        threadPoolExecutor= ThreadPoolUtil.getThreadPoolExecutor();
    }

    @Override
    public void asyncInvoke(T input, ResultFuture<T> resultFuture) {
        threadPoolExecutor.execute(new Runnable() {
            @SneakyThrows
            @Override
            public void run() {

                DruidPooledConnection phoenixConn = DruidDSUtil_hjy.getPhoenixConn();
                String id = getId(input);
                String filterId = getFilterId();
                JSONObject dimInfo = DimInfoFunction.getDimInfo(phoenixConn, tableName,filterId ,id);
                getDimInfo(input,dimInfo);
                resultFuture.complete(Collections.singleton(input));
                phoenixConn.close();
            }
        });

    }
    @Override
    public void timeout(T input, ResultFuture<T> resultFuture) throws Exception {
        System.out.println("超时数据："+input);
    }


}
